package drds.data_propagate.binlog;

import drds.data_propagate.binlog.event.*;
import drds.data_propagate.binlog.event.mariadb.*;
import drds.data_propagate.binlog.event.rows_event.DeleteRowsEvent;
import drds.data_propagate.binlog.event.rows_event.RowsEvent;
import drds.data_propagate.binlog.event.rows_event.UpdateRowsEvent;
import drds.data_propagate.binlog.event.rows_event.WriteRowsEvent;
import drds.data_propagate.driver.packets.GtidSet;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.io.IOException;
import java.util.BitSet;

/**
 * Implements a binary-log decoder.
 *
 * <pre>
 * Decoder decoder = new Decoder();
 * decoder.handle(...);
 *
 * BinLogEvent event;
 * do
 * {
 *     event = decoder.decode(bytes, context);
 *
 *     // process log event.
 * }
 * while (event != null);
 * // no more events in bytes.
 * </pre>
 */
public final class Decoder {

    protected static final Log logger = LogFactory.getLog(Decoder.class);

    protected final BitSet handleSet = new BitSet(BinLogEvent.ENUM_END_EVENT);

    public Decoder() {
    }

    public Decoder(final int fromIndex, final int toIndex) {
        handleSet.set(fromIndex, toIndex);
    }

    /**
     * Deserialize an event from bytes.
     *
     * @return <code>UknownLogEvent</code> if event eventType is unknown or skipped.
     */
    public static BinLogEvent decode(Buffer buffer, Header header, BinLogContext binLogContext) throws IOException {
        FormatDescriptionEvent formatDescriptionEvent = binLogContext.getFormatDescriptionEvent();
        BinlogPosition binlogPosition = binLogContext.getBinlogPosition();

        int checksumAlg = BinLogEvent.BINLOG_CHECKSUM_ALG_UNDEF;
        if (header.getEventType() != BinLogEvent.FORMAT_DESCRIPTION_EVENT) {
            checksumAlg = formatDescriptionEvent.header.getChecksumAlg();
        } else {
            // 如果是format事件自己，也需要处理checksum
            checksumAlg = header.getChecksumAlg();
        }

        if (checksumAlg != BinLogEvent.BINLOG_CHECKSUM_ALG_OFF && checksumAlg != BinLogEvent.BINLOG_CHECKSUM_ALG_UNDEF) {
            // remove checksum bytes
            buffer.limit(header.getEventLength() - BinLogEvent.BINLOG_CHECKSUM_LEN);
        }
        GtidSet gtidSet = binLogContext.getGtidSet();
        GtidEvent gtidEvent = binLogContext.getGtidEvent();
        switch (header.getEventType()) {
            case BinLogEvent.QUERY_EVENT: {
                QueryEvent queryEvent = new QueryEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                header.putGtid(binLogContext.getGtidSet(), gtidEvent);
                return queryEvent;
            }
            case BinLogEvent.XID_EVENT: {
                XidEvent xidEvent = new XidEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                header.putGtid(binLogContext.getGtidSet(), gtidEvent);
                return xidEvent;
            }
            case BinLogEvent.TABLE_MAP_EVENT: {
                TableMapEvent tableMapEvent = new TableMapEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                binLogContext.putTable(tableMapEvent);
                return tableMapEvent;
            }
            case BinLogEvent.WRITE_ROWS_EVENT_V1: {
                RowsEvent rowsEvent = new WriteRowsEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                rowsEvent.fillTable(binLogContext);
                header.putGtid(binLogContext.getGtidSet(), gtidEvent);
                return rowsEvent;
            }
            case BinLogEvent.UPDATE_ROWS_EVENT_V1: {
                RowsEvent rowsEvent = new UpdateRowsEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                rowsEvent.fillTable(binLogContext);
                header.putGtid(binLogContext.getGtidSet(), gtidEvent);
                return rowsEvent;
            }
            case BinLogEvent.DELETE_ROWS_EVENT_V1: {
                RowsEvent rowsEvent = new DeleteRowsEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                rowsEvent.fillTable(binLogContext);
                header.putGtid(binLogContext.getGtidSet(), gtidEvent);
                return rowsEvent;
            }
            case BinLogEvent.ROTATE_EVENT: {
                RotateEvent rotateEvent = new RotateEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition = new BinlogPosition(rotateEvent.getFilename(), rotateEvent.getPosition());
                binLogContext.setBinlogPosition(binlogPosition);
                return rotateEvent;
            }
            case BinLogEvent.LOAD_EVENT:
            case BinLogEvent.NEW_LOAD_EVENT: {
                LoadEvent loadEvent = new LoadEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                return loadEvent;
            }
            case BinLogEvent.SLAVE_EVENT: /* can never happen (unused event) */ {
                if (logger.isWarnEnabled()) logger.warn("Skipping unsupported SLAVE_EVENT from: "
                        + binLogContext.getBinlogPosition());
                break;
            }
            case BinLogEvent.CREATE_FILE_EVENT: {
                CreateFileEvent createFileEvent = new CreateFileEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                return createFileEvent;
            }
            case BinLogEvent.APPEND_BLOCK_EVENT: {
                AppendBlockEvent appendBlockEvent = new AppendBlockEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                return appendBlockEvent;
            }
            case BinLogEvent.DELETE_FILE_EVENT: {
                DeleteFileEvent deleteFileEvent = new DeleteFileEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                return deleteFileEvent;
            }
            case BinLogEvent.EXEC_LOAD_EVENT: {
                ExecuteLoadEvent executeLoadEvent = new ExecuteLoadEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                return executeLoadEvent;
            }
            case BinLogEvent.START_EVENT_V3: {
                /* This is sent only by MySQL <=4.x */
                StartEventV3 startEventV3 = new StartEventV3(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                return startEventV3;
            }
            case BinLogEvent.STOP_EVENT: {
                StopEvent stopEvent = new StopEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                return stopEvent;
            }
            case BinLogEvent.INTVAR_EVENT: {
                IntvarEvent intvarEvent = new IntvarEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                return intvarEvent;
            }
            case BinLogEvent.RAND_EVENT: {
                RandEvent randEvent = new RandEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                header.putGtid(binLogContext.getGtidSet(), gtidEvent);
                return randEvent;
            }
            case BinLogEvent.USER_VAR_EVENT: {
                UserVarEvent userVarEvent = new UserVarEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                header.putGtid(binLogContext.getGtidSet(), gtidEvent);
                return userVarEvent;
            }
            case BinLogEvent.FORMAT_DESCRIPTION_EVENT: {
                formatDescriptionEvent = new FormatDescriptionEvent(header, buffer, formatDescriptionEvent);
                binLogContext.setFormatDescriptionEvent(formatDescriptionEvent);
                return formatDescriptionEvent;
            }
            case BinLogEvent.PRE_GA_WRITE_ROWS_EVENT: {
                if (logger.isWarnEnabled()) logger.warn("Skipping unsupported PRE_GA_WRITE_ROWS_EVENT from: "
                        + binLogContext.getBinlogPosition());
                // ev = new Write_rows_log_event_old(buf, event_len,
                // description_event);
                break;
            }
            case BinLogEvent.PRE_GA_UPDATE_ROWS_EVENT: {
                if (logger.isWarnEnabled()) logger.warn("Skipping unsupported PRE_GA_UPDATE_ROWS_EVENT from: "
                        + binLogContext.getBinlogPosition());
                // ev = new Update_rows_log_event_old(buf, event_len,
                // description_event);
                break;
            }
            case BinLogEvent.PRE_GA_DELETE_ROWS_EVENT: {
                if (logger.isWarnEnabled()) logger.warn("Skipping unsupported PRE_GA_DELETE_ROWS_EVENT from: "
                        + binLogContext.getBinlogPosition());
                // ev = new Delete_rows_log_event_old(buf, event_len,
                // description_event);
                break;
            }
            case BinLogEvent.BEGIN_LOAD_QUERY_EVENT: {
                BeginLoadQueryEvent beginLoadQueryEvent = new BeginLoadQueryEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                return beginLoadQueryEvent;
            }
            case BinLogEvent.EXECUTE_LOAD_QUERY_EVENT: {
                ExecuteLoadQueryEvent executeLoadQueryEvent = new ExecuteLoadQueryEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                return executeLoadQueryEvent;
            }
            case BinLogEvent.INCIDENT_EVENT: {
                IncidentEvent incidentEvent = new IncidentEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                return incidentEvent;
            }
            case BinLogEvent.HEARTBEAT_LOG_EVENT: {
                HeartbeatEvent heartbeatEvent = new HeartbeatEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                return heartbeatEvent;
            }
            case BinLogEvent.IGNORABLE_LOG_EVENT: {
                IgnorableEvent ignorableEvent = new IgnorableEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                return ignorableEvent;
            }
            case BinLogEvent.ROWS_QUERY_LOG_EVENT: {
                RowsQueryEvent rowsQueryEvent = new RowsQueryEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                header.putGtid(binLogContext.getGtidSet(), gtidEvent);
                return rowsQueryEvent;
            }
            case BinLogEvent.WRITE_ROWS_EVENT: {
                RowsEvent writeRowsEvent = new WriteRowsEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                writeRowsEvent.fillTable(binLogContext);
                header.putGtid(binLogContext.getGtidSet(), gtidEvent);
                return writeRowsEvent;
            }
            case BinLogEvent.UPDATE_ROWS_EVENT: {
                RowsEvent rowsEvent = new UpdateRowsEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                rowsEvent.fillTable(binLogContext);
                header.putGtid(binLogContext.getGtidSet(), gtidEvent);
                return rowsEvent;
            }
            case BinLogEvent.DELETE_ROWS_EVENT: {
                RowsEvent rowsEvent = new DeleteRowsEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                rowsEvent.fillTable(binLogContext);
                header.putGtid(binLogContext.getGtidSet(), gtidEvent);
                return rowsEvent;
            }
            case BinLogEvent.PARTIAL_UPDATE_ROWS_EVENT: {
                RowsEvent rowsEvent = new UpdateRowsEvent(header, buffer, formatDescriptionEvent, true);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                rowsEvent.fillTable(binLogContext);
                header.putGtid(binLogContext.getGtidSet(), gtidEvent);
                return rowsEvent;
            }
            case BinLogEvent.GTID_LOG_EVENT:
            case BinLogEvent.ANONYMOUS_GTID_LOG_EVENT: {
                GtidEvent gtidEvent1 = new GtidEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                if (gtidSet != null) {
                    gtidSet.update(gtidEvent1.getGtidStr());
                    // update latest gtid
                    header.putGtid(gtidSet, gtidEvent1);
                }
                // update current gtid event to context
                binLogContext.setGtidEvent(gtidEvent1);
                return gtidEvent1;
            }
            case BinLogEvent.PREVIOUS_GTIDS_LOG_EVENT: {
                PreviousGtidsEvent previousGtidsEvent = new PreviousGtidsEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                return previousGtidsEvent;
            }
            case BinLogEvent.TRANSACTION_CONTEXT_EVENT: {
                TransactionContextEvent transactionContextEvent = new TransactionContextEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                return transactionContextEvent;
            }
            case BinLogEvent.VIEW_CHANGE_EVENT: {
                ViewChangeEvent viewChangeEvent = new ViewChangeEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                return viewChangeEvent;
            }
            case BinLogEvent.XA_PREPARE_LOG_EVENT: {
                XaPrepareEvent xaPrepareEvent = new XaPrepareEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                return xaPrepareEvent;
            }
            case BinLogEvent.ANNOTATE_ROWS_EVENT: {
                AnnotateRowsEvent annotateRowsEvent = new AnnotateRowsEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                header.putGtid(binLogContext.getGtidSet(), gtidEvent);
                return annotateRowsEvent;
            }
            case BinLogEvent.BINLOG_CHECKPOINT_EVENT: {
                BinlogCheckPointEvent binlogCheckPointEvent = new BinlogCheckPointEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                return binlogCheckPointEvent;
            }
            case BinLogEvent.GTID_EVENT: {
                MariaGtidEvent mariaGtidEvent = new MariaGtidEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                return mariaGtidEvent;
            }
            case BinLogEvent.GTID_LIST_EVENT: {
                MariaGtidListEvent mariaGtidListEvent = new MariaGtidListEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                return mariaGtidListEvent;
            }
            case BinLogEvent.START_ENCRYPTION_EVENT: {
                StartEncryptionEvent startEncryptionEvent = new StartEncryptionEvent(header, buffer, formatDescriptionEvent);
                /* updating position in context */
                binlogPosition.position = header.getLogPosition();
                return startEncryptionEvent;
            }
            default:
                /*
                 * Create an object of Ignorable_log_event for unrecognized
                 * sub-class. So that SLAVE SQL THREAD will only update the
                 * position and continue.
                 */
                if ((buffer.getUint16(BinLogEvent.FLAGS_OFFSET) & BinLogEvent.LOG_EVENT_IGNORABLE_F) > 0) {
                    IgnorableEvent ignorableEvent = new IgnorableEvent(header, buffer, formatDescriptionEvent);
                    /* updating position in context */
                    binlogPosition.position = header.getLogPosition();
                    return ignorableEvent;
                } else {
                    if (logger.isWarnEnabled()) {
                        logger.warn("Skipping unrecognized binlog event " + BinLogEvent.getTypeName(header.getEventType())
                                + " from: " + binLogContext.getBinlogPosition());
                    }
                }
        }

        /* updating position in context */
        binlogPosition.position = header.getLogPosition();
        /* Unknown or unsupported log event */
        return new UnknownEvent(header);
    }

    public final void handle(final int fromIndex, final int toIndex) {
        handleSet.set(fromIndex, toIndex);
    }

    public final void handle(final int flagIndex) {
        handleSet.set(flagIndex);
    }

    /**
     * Decoding an event from binary-log bytes.
     *
     * @return <code>UknownLogEvent</code> if event eventType is unknown or skipped,
     * <code>null</code> if bytes is not including a full event.
     */
    public BinLogEvent decode(Buffer buffer, BinLogContext binLogContext) throws IOException {
        final int limit = buffer.limit();

        if (limit >= FormatDescriptionEvent.LOG_EVENT_HEADER_LEN) {
            Header header = new Header(buffer, binLogContext.getFormatDescriptionEvent());

            final int len = header.getEventLength();
            if (limit >= len) {
                BinLogEvent event;

                /* Checking binary-log's headerPacket */
                if (handleSet.get(header.getEventType())) {
                    buffer.limit(len);
                    try {
                        /* Decoding binary-log to event */
                        event = decode(buffer, header, binLogContext);
                    } catch (IOException e) {
                        if (logger.isWarnEnabled()) {
                            logger.warn("Decoding " + BinLogEvent.getTypeName(header.getEventType()) + " failed from: "
                                    + binLogContext.getBinlogPosition(), e);
                        }
                        throw e;
                    } finally {
                        buffer.limit(limit); /* Restore limit */
                    }
                } else {
                    /* Ignore unsupported binary-log. */
                    event = new UnknownEvent(header);
                }

                if (event != null) {
                    // set logFileName
                    event.getHeader().setLogFileName(binLogContext.getBinlogPosition().getFileName());
                    event.setSemival(buffer.semival);
                }

                /* consume this binary-log. */
                buffer.consume(len);
                return event;
            }
        }

        /* Rewind bytes's position to 0. */
        buffer.rewind();
        return null;
    }
}
